package com.gitee.wsl.func.filter.bloomfilter

import com.gitee.wsl.ext.base.Arrays.copyOf
import com.gitee.wsl.func.filter.bloomfilter.api.BitMapExtractor
import com.gitee.wsl.func.filter.bloomfilter.api.BloomFilter
import com.gitee.wsl.func.filter.bloomfilter.api.BloomFilterExtractor
import com.gitee.wsl.func.filter.bloomfilter.api.Hasher
import com.gitee.wsl.func.filter.bloomfilter.api.IndexExtractor
import com.gitee.wsl.func.filter.bloomfilter.bean.SetOperations
import com.gitee.wsl.func.filter.bloomfilter.bean.Shape
import com.gitee.wsl.func.filter.bloomfilter.extractor.LayerManager
import com.gitee.wsl.func.predicate.IntPredicate
import com.gitee.wsl.func.predicate.LongPredicate
import com.gitee.wsl.func.predicate.Predicate



/**
 * Layered Bloom filters are described in Zhiwang, Cen; Jungang, Xu; Jian, Sun (2010), "A multi-layer Bloom filter for duplicated URL detection", Proc. 3rd
 * International Conference on Advanced Computer Theory and Engineering (ICACTE 2010), vol. 1, pp. V1-586-V1-591, doi:10.1109/ICACTE.2010.5578947, ISBN
 * 978-1-4244-6539-2, S2CID 3108985
 *
 *
 * In short, Layered Bloom filter contains several bloom filters arranged in layers.
 *
 *
 *  * When membership in the filter is checked each layer in turn is checked and if a match is found `true` is returned.
 *  * When merging each bloom filter is merged into the newest filter in the list of layers.
 *  * When questions of cardinality are asked the cardinality of the union of the enclosed Bloom filters is used.
 *
 *
 *
 * The net result is that the layered Bloom filter can be populated with more items than the Shape would indicate and yet still return a false positive rate in
 * line with the Shape and not the over population.
 *
 *
 *
 * This implementation uses a LayerManager to handle the manipulation of the layers.
 *
 *
 *  * Level 0 is the oldest layer and the highest level is the newest.
 *  * There is always at least one enclosed filter.
 *  * The newest filter is the `target` into which merges are performed.
 *  * Whenever the target is retrieved, or a `merge` operation is performed the code checks if any older layers should be removed, and if so removes
 * them. It also checks it a new layer should be added, and if so adds it and sets the `target` before the operation.
 *
 *
 * @param <T> The type of Bloom Filter that is used for the layers.
 * @since 4.5.0-M2
</T> */
class LayeredBloomFilter<T : BloomFilter<T>>(
    override val shape: Shape,
    private val layerManager: LayerManager<T>
) : BloomFilter<LayeredBloomFilter<T>>, BloomFilterExtractor<T> {
    /**
     * A class used to locate matching filters across all the layers.
     */
    private inner class Finder(var bf: T) : Predicate<T> {
        private var result: IntArray = IntArray(layerManager.depth)
        var bfIdx: Int = 0
        var resultIdx: Int = 0

        fun getResult(): IntArray {
            return copyOf(result, resultIdx)
        }

        override fun test(t: T): Boolean {
            if (t.contains(bf)) {
                result[resultIdx++] = bfIdx
            }
            bfIdx++
            return true
        }
    }

     override fun cardinality(): Int {
        return SetOperations.cardinality(this)
    }

     override fun characteristics(): Int {
        return 0
    }

    /**
     * Forces the execution of the cleanup Consumer that was provided when the associated LayerManager was built.
     *
     * @see LayerManager.Builder.setCleanup
     */
    fun cleanup() {
        layerManager.cleanup()
    }

     override fun clear() {
        layerManager.clear()
    }

     override fun contains(bitMapExtractor: BitMapExtractor): Boolean {
        return contains(createFilter(bitMapExtractor))
    }

    /**
     * Returns `true` if this any layer contained by this filter contains the specified filter.
     *
     *
     * If the `other` is a BloomFilterExtractor each filter within the `other` is checked to see if it exits within this filter.
     *
     *
     * @param other the other Bloom filter
     * @return `true` if this filter contains the other filter.
     */
    override fun contains(other: BloomFilter<*>): Boolean {
        return if (other is BloomFilterExtractor<*>)
                   contains(other as BloomFilterExtractor<T>)
               else
                   !processBloomFilters(Predicate { x -> !x.contains(other) })
    }

    /**
     * Returns `true` if each filter within the `bloomFilterExtractor` exits within this filter.
     *
     * @param bloomFilterExtractor the BloomFilterExtractor that provides the filters to check for.
     * @return `true` if this filter contains all of the filters contained in the `bloomFilterExtractor`.
     */
    fun contains(bloomFilterExtractor: BloomFilterExtractor<T>): Boolean {
        val result = booleanArrayOf(true)
        // return false when we have found a match to short circuit checks
        return bloomFilterExtractor.processBloomFilters(Predicate { x ->
            result[0] = result[0] and contains(x)
            result[0]
        })
    }

     override fun contains(hasher: Hasher): Boolean {
        return contains(createFilter(hasher))
    }

     override fun contains(indexExtractor: IndexExtractor): Boolean {
        return contains(createFilter(indexExtractor))
    }

    /**
     * Creates a new instance of this [LayeredBloomFilter] with the same properties as the current one.
     *
     * @return a copy of this [LayeredBloomFilter].
     */
     override fun copy(): LayeredBloomFilter<T> {
        return LayeredBloomFilter(shape, layerManager.copy())
    }

    /**
     * Creates a Bloom filter from a BitMapExtractor.
     *
     * @param bitMapExtractor the BitMapExtractor to create the filter from.
     * @return the BloomFilter.
     */
    private fun createFilter(bitMapExtractor: BitMapExtractor): SimpleBloomFilter {
        val bf = SimpleBloomFilter(shape)
        bf.merge(bitMapExtractor)
        return bf
    }

    /**
     * Creates a Bloom filter from a Hasher.
     *
     * @param hasher the hasher to create the filter from.
     * @return the BloomFilter.
     */
    private fun createFilter(hasher: Hasher): SimpleBloomFilter {
        val bf = SimpleBloomFilter(shape)
        bf.merge(hasher)
        return bf
    }

    /**
     * Creates a Bloom filter from an IndexExtractor.
     *
     * @param indexExtractor the IndexExtractor to create the filter from.
     * @return the BloomFilter.
     */
    private fun createFilter(indexExtractor: IndexExtractor): SimpleBloomFilter {
        val bf = SimpleBloomFilter(shape)
        bf.merge(indexExtractor)
        return bf
    }

     override fun estimateN(): Int {
        return flatten().estimateN()
    }

     override fun estimateUnion(other: BloomFilter<*>): Int {
        val cpy = this.flatten()
        cpy.merge(other)
        return cpy.estimateN()
    }

    /**
     * Finds the layers in which the BitMapExtractor is found.
     *
     * @param bitMapExtractor the BitMapExtractor to search for.
     * @return an array of layer indices in which the Bloom filter is found.
     */
    fun find(bitMapExtractor: BitMapExtractor): IntArray? {
        val bf = SimpleBloomFilter(shape)
        bf.merge(bitMapExtractor)
        return find(bf)
    }

    /**
     * Finds the layers in which the Bloom filter is found.
     *
     * @param bf the Bloom filter to search for.
     * @return an array of layer indices in which the Bloom filter is found.
     */
    fun find(bf: BloomFilter<*>): IntArray {
        val finder = Finder(bf as T)
        processBloomFilters(finder)
        return finder.getResult()
    }

    /**
     * Finds the layers in which the Hasher is found.
     *
     * @param hasher the Hasher to search for.
     * @return an array of layer indices in which the Bloom filter is found.
     */
    fun find(hasher: Hasher): IntArray {
        val bf = SimpleBloomFilter(shape)
        bf.merge(hasher)
        return find(bf)
    }

    /**
     * Finds the layers in which the IndexExtractor is found.
     *
     * @param indexExtractor the Index extractor to search for.
     * @return an array of layer indices in which the Bloom filter is found.
     */
    fun find(indexExtractor: IndexExtractor): IntArray {
        val bf = SimpleBloomFilter(shape)
        bf.merge(indexExtractor)
        return find(bf)
    }

    /**
     * Create a standard (non-layered) Bloom filter by merging all of the layers. If the filter is empty this method will return an empty Bloom filter.
     *
     * @return the merged bloom filter.
     */
    override fun flatten(): BloomFilter<T> {
        val bf = SimpleBloomFilter(shape)
        processBloomFilters(bf::merge)
        return bf as BloomFilter<T>
    }

    /**
     * Gets the Bloom filter at the specified depth
     *
     * @param depth the depth of the filter to return.
     * @return the Bloom filter at the specified depth.
     * @throws NoSuchElementException if depth is not in the range [0,getDepth())
     */
    fun get(depth: Int): T? {
        return layerManager.get(depth)
    }

    val depth: Int
        /**
         * Gets the depth of the deepest layer. The minimum value returned by this method is 1.
         *
         * @return the depth of the deepest layer.
         */
        get() = layerManager.depth


    override fun isEmpty(): Boolean {
        return processBloomFilters(BloomFilter<T>::isEmpty)
    }

     override fun merge(bitMapExtractor: BitMapExtractor): Boolean {
        return layerManager.target!!.merge(bitMapExtractor)
    }

     override fun merge(other: BloomFilter<*>): Boolean {
        return layerManager.target!!.merge(other)
    }

     override fun merge(indexExtractor: IndexExtractor): Boolean {
        return layerManager.target!!.merge(indexExtractor)
    }

    /**
     * Forces and advance to the next layer. This method will clean-up the current layers and generate a new filter layer. In most cases is it unnecessary to
     * call this method directly.
     *
     * @see LayerManager.Builder.setCleanup
     * @see LayerManager.Builder.setExtendCheck
     */
    fun next() {
        layerManager.next()
    }

     override fun processBitMaps(predicate: LongPredicate): Boolean {
        return flatten().processBitMaps(predicate)
    }

    /**
     * Processes the Bloom filters in depth order with the most recent filters first. Each filter is passed to the predicate in turn. The function exits on the
     * first `false` returned by the predicate.
     *
     * @param bloomFilterPredicate the predicate to execute.
     * @return `true` if all filters passed the predicate, `false` otherwise.
     */
     override fun processBloomFilters(bloomFilterPredicate: Predicate<T>): Boolean {
        return layerManager.processBloomFilters(bloomFilterPredicate)
    }

     override fun processIndices(predicate: IntPredicate): Boolean {
        return processBloomFilters(Predicate { bf -> bf.processIndices(predicate) })
    }
}
